/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nageoffer.shortlink.project.config;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.hash.MurmurHash;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.project.dao.entity.ShortLinkDO;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkMapper;
import com.nageoffer.shortlink.project.toolkit.BloomFilterIdxUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.math.BigDecimal;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 布隆过滤器配置
 */
@Slf4j
@Configuration
@RequiredArgsConstructor
public class RBloomFilterConfiguration implements CommandLineRunner {

    private final ShortLinkMapper shortLinkMapper;
    private final StringRedisTemplate stringRedisTemplate;
    private final RedissonClient redissonClient;

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    private Integer calculateExpansionSize(int size) {
        return new BigDecimal(size).multiply(new BigDecimal("1.5")).intValue();
    }

    /**
     * 防止短链接创建查询数据库的布隆过滤器
     *
     * 该方法旨在初始化并提供一个布隆过滤器实例，用于缓存穿透的处理
     * 缓存穿透是指大量请求访问数据库中不存在的数据，导致数据库压力剧增的情况
     * 使用布隆过滤器可以有效避免此类问题，因为它可以快速判断数据是否存在，而无需查询数据库
     *
     * @param redissonClient Redis客户端，用于操作Redis数据库
     * @return RBloomFilter<String> 返回一个布隆过滤器实例，用于缓存穿透的处理
     */
    @Bean
    public RBloomFilter<String> shortUriCreateCachePenetrationBloomFilter(RedissonClient redissonClient) {
        // 获取布隆过滤器实例
        RBloomFilter<String> cachePenetrationBloomFilter = redissonClient.getBloomFilter("shortUriCreateCachePenetrationBloomFilter");
        // 尝试初始化布隆过滤器，参数分别为预计插入的元素数量和期望的误算率
        cachePenetrationBloomFilter.tryInit(100000000L, 0.001);
        // 返回初始化后的布隆过滤器实例
        return cachePenetrationBloomFilter;
    }

    @Override
    public void run(String... args) throws Exception {
        String bloomFilterSizeStr = stringRedisTemplate.opsForValue().get("sharding-expansion:bloom_filter_size");
        if (StrUtil.isEmpty(bloomFilterSizeStr)) {
            bloomFilterSizeStr = "2";
            stringRedisTemplate.opsForValue().set("sharding-expansion:bloom_filter_size", bloomFilterSizeStr);
        }

        // 使用 int 也行，不过因为 lambda 中不允许使用变化的常量，所以这里使用 AtomicInteger 包装一层
        AtomicInteger bloomFilterSize = new AtomicInteger(Integer.parseInt(bloomFilterSizeStr));

        String bloomFilterCurrentIndex = stringRedisTemplate.opsForValue().get("sharding-expansion:bloom_filter_current_index");
        if (StrUtil.isEmpty(bloomFilterCurrentIndex)) {
            stringRedisTemplate.opsForValue().set("sharding-expansion:bloom_filter_current_index", "0");
        }
        bloomFilterCurrentIndex = bloomFilterCurrentIndex == null ? "0" : bloomFilterCurrentIndex;
        for (int i = 0; i < bloomFilterSize.get(); i++) {
            RBloomFilter<String> bloomFilter = redissonClient.getBloomFilter(String.format("sharding-expansion:short-link-create-bloom-filter-%s-%d", bloomFilterCurrentIndex, i));
            bloomFilter.tryInit(6L, 0.001);
        }

        // 分布式环境下再加个分布式锁，下面的检查需要是单机执行的
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            String scheduledBloomFilterCurrentIndex = stringRedisTemplate.opsForValue().get("sharding-expansion:bloom_filter_current_index");
            String expansionFlag = stringRedisTemplate.opsForValue().get("sharding-expansion:expansion_flag");
            if (StrUtil.isNotBlank(expansionFlag) && expansionFlag.equals("true")) {
                return;
            }

            String actualBloomFilterCurrentIndex = stringRedisTemplate.opsForValue().get("sharding-expansion:bloom_filter_current_index");
            int originalBloomFilterSize = bloomFilterSize.get();
            for (int i = 0; i < originalBloomFilterSize; i++) {
                String currentBloomFilterName = String.format("sharding-expansion:short-link-create-bloom-filter-%s-%d", actualBloomFilterCurrentIndex, i);
                RBloomFilter<String> bloomFilter = redissonClient.getBloomFilter(currentBloomFilterName);
                long count = bloomFilter.count();
                // 任意一个超过布隆过滤器通超过 80% 意味着需要扩容
                long initId = 0;
                if (count > 4) {
                    log.info("触发布隆过滤器扩容操作，当前布隆过滤器名称：{}", currentBloomFilterName);
                    stringRedisTemplate.opsForValue().set("sharding-expansion:expansion_flag", "true");
                    String expansionIdx = BloomFilterIdxUtil.reversal(scheduledBloomFilterCurrentIndex);

                    // 初始化扩容后的布隆过滤器
                    Integer expansionBloomFilterSize = calculateExpansionSize(originalBloomFilterSize);
                    for (int j = 0; j < expansionBloomFilterSize; j++) {
                        RBloomFilter<String> expansionBloomFilter = redissonClient.getBloomFilter(String.format("sharding-expansion:short-link-create-bloom-filter-%s-%d", expansionIdx, j));
                        expansionBloomFilter.tryInit(12L, 0.001);
                    }

                    bloomFilterSize.set(expansionBloomFilterSize);

                    // 查询数据库最大的短链接记录表 ID，作为扩容的上限
                    LambdaQueryWrapper<ShortLinkDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkDO.class)
                            .select(ShortLinkDO::getId)
                            .orderByDesc(ShortLinkDO::getId)
                            .last("LIMIT 1");
                    ShortLinkDO shortLinkDO = shortLinkMapper.selectOne(queryWrapper);

                    while (true) {
                        queryWrapper = Wrappers.lambdaQuery(ShortLinkDO.class)
                                .gt(ShortLinkDO::getId, initId)
                                .le(ShortLinkDO::getId, shortLinkDO.getId())
                                .last("LIMIT 2");
                        List<ShortLinkDO> shortLinkDOList = shortLinkMapper.selectList(queryWrapper);

                        for (ShortLinkDO each : shortLinkDOList) {
                            int hash32 = MurmurHash.hash32(each.getFullShortUrl());
                            long mod = hash32 < 0 ? Integer.MAX_VALUE - (long) hash32 : hash32;
                            long idx = mod % expansionBloomFilterSize;
                            log.info("当前短链接 {} 索引下标位置：{}", each.getFullShortUrl(), idx);

                            RBloomFilter<String> expansionBloomFilter = redissonClient.getBloomFilter(String.format("sharding-expansion:short-link-create-bloom-filter-%s-%d", expansionIdx, idx));
                            expansionBloomFilter.add(each.getFullShortUrl());
                        }

                        if (CollUtil.isEmpty(shortLinkDOList) || shortLinkDOList.size() < 2) {
                            log.info("当前短链接布隆过滤器扩容初步完成...");
                            break;
                        }

                        initId = shortLinkDOList.stream()
                                .mapToLong(ShortLinkDO::getId)
                                .max()
                                .orElse(initId);
                    }

                    // 扩容完成后设置新的布隆过滤器标识，同时设置扩容标识为 false
                    // 正常来说将这些逻辑需要使用 Lua 脚本执行，确保同时执行
                    stringRedisTemplate.opsForValue().set("sharding-expansion:bloom_filter_current_index", expansionIdx);
                    stringRedisTemplate.opsForValue().set("sharding-expansion:expansion_flag", "false");
                    stringRedisTemplate.opsForValue().set("sharding-expansion:bloom_filter_size", expansionBloomFilterSize.toString());

                    // 如果在咱们上面 while 循环最后一次执行后，到设置扩容后的布隆过滤器标识这个时间段有新增的短链接
                    // 再执行一次同步，为了避免在扩容时创建的短链接，这是一个教小的时间周期，但是也需要考虑
                    while (true) {
                        queryWrapper = Wrappers.lambdaQuery(ShortLinkDO.class)
                                .gt(ShortLinkDO::getId, initId)
                                .le(ShortLinkDO::getId, shortLinkDO.getId())
                                .last("LIMIT 2");
                        List<ShortLinkDO> shortLinkDOList = shortLinkMapper.selectList(queryWrapper);

                        for (ShortLinkDO each : shortLinkDOList) {
                            int hash32 = MurmurHash.hash32(each.getFullShortUrl());
                            long mod = hash32 < 0 ? Integer.MAX_VALUE - (long) hash32 : hash32;
                            long idx = mod % expansionBloomFilterSize;
                            log.info("当前短链接 {} 索引下标位置：{}", each.getFullShortUrl(), idx);

                            RBloomFilter<String> expansionBloomFilter = redissonClient.getBloomFilter(String.format("sharding-expansion:short-link-create-bloom-filter-%s-%d", expansionIdx, idx));
                            expansionBloomFilter.add(each.getFullShortUrl());
                        }

                        if (CollUtil.isEmpty(shortLinkDOList) || shortLinkDOList.size() < 2) {
                            log.info("当前短链接布隆过滤器扩容已完成...");
                            break;
                        }

                        initId = shortLinkDOList.stream()
                                .mapToLong(ShortLinkDO::getId)
                                .max()
                                .orElse(initId);
                    }

                    // 删除原布隆过滤器
                    for (int j = 0; j < originalBloomFilterSize; j++) {
                        RBloomFilter<String> expansionBloomFilter = redissonClient.getBloomFilter(String.format("sharding-expansion:short-link-create-bloom-filter-%s-%d", scheduledBloomFilterCurrentIndex, j));
                        expansionBloomFilter.delete();
                    }
                    break;
                }
            }
        }, 0, 10, TimeUnit.SECONDS);
    }
}




